#include "function.h"

//连接mqtt_server
MQTTClient client;          // 声明一个MQTT客户端对象
char addr[32];              // mqtt broker ip地址
int port;                   // 端口号

sqlite3 *db;                // 数据库句柄
int stm32_num = 0;          // stm32设备数
int modbus_num = 0;         // modbus设备数
int sum = 0;                // 设备点总个数

int sbtype = 1;             // 上报类型---默认为变化上报
int zltype = 0;             // 指令类型
int period;                 // 轮询周期

int keys_part[32] = {0};    // 刷新上报设备点key值数组
char *limit;                // 采集数据点范围

int key_num;                // 存储设备键值
int limit_head, limit_tail; // 存储时间戳范围

int key_ctrl;               // 要控制的设备key值
char *val_ctrl;             // 要修改成的值

// 共享内存、消息队列
int shmid;                 // 共享内存ID
int msgid;                 // 消息队列ID
struct std_node *shm_data; // 结构体指针

// 前置定义
int shm_init(size_t shm_size,char *name); // 创建共享内存
void *shm_getaddr();           // 获取共享内存地址
int periodic_reporting();      // 轮询上报
int select_history();          // 查询指令
void mqtt_send(char *sendbuf); //mqtt发送函数
// 字符转换函数
// 将浮点数转换为字符串
char *f_to_str(float value)
{
    static char str[50];
    snprintf(str, sizeof(str), "%.2f", value); // 保留两位小数
    return str;
}
// 将整数转换为字符串
char *i_to_str(int value)
{
    static char str[50];
    snprintf(str, sizeof(str), "%d", value);
    return str;
}
// 解析数据点表
int parse_point_table()
{

    // 创建cJson节点
    int fd = open(NODE_PATH, O_RDONLY);
    if (fd < 0)
    {
        perror("open err");
        return -1;
    }
    char buf[4096] = {0};
    int ret = read(fd, buf, 1024);
    if (ret < 0)
    {
        perror("read error");
    }
    close(fd); // 读取完后关闭文件
    // 把传入的cJSON字符串转成cJSON的结构
    cJSON *root = cJSON_Parse(buf);
    if (NULL == root)
    {
        printf("cJSON_Parse error\n");
        return -1;
    }
    // 解析mqtt_server
    cJSON *mqtt_server = cJSON_GetObjectItem(root, "mqtt_server");
    // addr
    cJSON *tmp = cJSON_GetObjectItem(mqtt_server, "addr");
    strcpy(addr, tmp->valuestring);
    // port
    tmp = cJSON_GetObjectItem(mqtt_server, "port");
    port = tmp->valueint;
    cJSON *stm32data = NULL;
    // 解析report type period
    cJSON *report = cJSON_GetObjectItem(root, "report");
    cJSON *report_type = cJSON_GetObjectItem(report, "type");
    sbtype = report_type->valueint;
    if (sbtype == 2)
    {
        cJSON *report_period = cJSON_GetObjectItem(report, "period");
        period = report_period->valueint;
    }
    // STM32
    cJSON *stm32 = cJSON_GetObjectItem(root, "stm32");
    if (stm32 == NULL)
    {
        printf("no such stm32\n");
        return -1;
    }
    else
    {
        stm32data = cJSON_GetObjectItem(stm32, "data");
        // 获取数组的大小
        stm32_num = cJSON_GetArraySize(stm32data);
        sum += stm32_num;
    }
    // modbus
    cJSON *modbus = cJSON_GetObjectItem(root, "modbus");
    if (modbus == NULL)
    {
        printf("no such modbus\n");
        return -1;
    }
    else
    {
        cJSON *modbusname = cJSON_GetObjectItem(modbus, "name");
        cJSON *modbusdata = cJSON_GetObjectItem(modbus, "data");
        // 获取数组的大小
        modbus_num = cJSON_GetArraySize(modbusdata);
        sum += modbus_num;
    }
    // 创建共享内存
    size_t shm_size = sum * sizeof(struct std_node) + sizeof(int);
    shm_init(shm_size,SHM_NAME);
    // 映射共享内存
    int *num = shm_getaddr();
    if (num == NULL)
    {
        printf("fuck\n");
        return -1;
    }
    // 写入总个数
    *num = sum;
    shm_data = (struct std_node *)(num + 1); //+4指向结构体数组的开始
    // 遍历共享内存
    // 写入结构体
    int count = 0;
    // 写入STM32设备点至共享内存
    for (int i = 0; i < stm32_num; i++)
    {
        // 从array数组中获取第item个子节点
        cJSON *item = cJSON_GetArrayItem(stm32data, i);
        //  //获取object大节点名字叫str
        cJSON *str = cJSON_GetObjectItem(item, "key");
        if (str == NULL)
        {
            return -1;
        }
        shm_data[i].key = str->valueint;
        str = cJSON_GetObjectItem(item, "type"); // 数据点数据类型
        if (str == NULL)
        {
            return -1;
        }
        shm_data[i].type = str->valueint;
        shm_data[i].dev_type = 101; // STM32设备类型
        // 新旧值
        switch (shm_data[i].type)
        {
        case 1: // 1-bool 2-int 3-float
            shm_data[i].new_val.b_val = 0;
            shm_data[i].old_val.b_val = 0;
            break;
        case 2:
            shm_data[i].new_val.i_val = 0;
            shm_data[i].old_val.i_val = 0;
            break;
        case 3:
            shm_data[i].new_val.f_val = 0;
            shm_data[i].old_val.f_val = 0;
            break;
        default:
            break;
        }
        shm_data[i].ret = -1;
        count++;
    }
    // 写入MODBUS设备点至共享内存
    for (int i = 0; i < modbus_num; i++)
    {

        cJSON *modbusdata = cJSON_GetObjectItem(modbus, "data");
        // 从array数组中获取第i个子节点

        cJSON *item = cJSON_GetArrayItem(modbusdata, i);
        // 获取object大节点名字叫str
        cJSON *str = cJSON_GetObjectItem(item, "key");
        if (str == NULL)
        {
            return -1;
        }

        shm_data[i + count].key = str->valueint;

        str = cJSON_GetObjectItem(item, "type");
        if (str == NULL)
        {
            return -1;
        }

        shm_data[i + count].type = str->valueint;

        shm_data[i + count].dev_type = 102; // Modbus设备类型
        // 新旧值
        switch (shm_data[i + count].type)
        {
        case 1: // 1-bool 2-int 3-float
            shm_data[i + count].new_val.b_val = 0;
            shm_data[i + count].old_val.b_val = 0;
            break;
        case 2:
            shm_data[i + count].new_val.i_val = 0;
            shm_data[i + count].old_val.i_val = 0;
            break;
        case 3:
            shm_data[i + count].new_val.f_val = 0;
            shm_data[i + count].old_val.f_val = 0;
            break;
        default:
            break;
        }
        shm_data[i + count].ret = -1;
    }
    // 释放节点
    cJSON_Delete(root);
}
//发送消息队列
int msg_queue_send(const char *name, const void *msg, size_t msgsz, int msgflg)
{
    assert(NULL != name && strlen(name) > 0);
    assert(NULL != msg);

    key_t key;
    int ret;
    int msgid;
    char sys_cmd[256];
    char path[128];

    sprintf(path, "%s%s", "/tmp/ipc/msgqueue/peer/", name);

    //文件不存在创建
    if(access(path, F_OK) < 0)
    {
        sprintf(sys_cmd, "%s %s", "touch", path);
        ret = system(sys_cmd);
        UNUSED(ret);
    }

    //创建key
    key = ftok(path, MAGIC_ID);
    if(key < 0){
        perror("fail to ftok");
        return -1;
    }

    //创建消息队列
    msgid = msgget(key, IPC_CREAT|0666);
    if (msgid < 0)
    {
        perror("fail to msgget");
        return -1;
    }

    return msgsnd(msgid, msg, msgsz - sizeof(long), msgflg);
}
// 创建共享内存
int shm_init(size_t shm_size,char *name)
{
    assert(NULL != name && strlen(name) > 0);
    key_t key;
    int ret;
    int id;
    char sys_cmd[256];
    char path[128];

    sprintf(path, "/tmp/ipc/shmem/%s", name);
    sprintf(sys_cmd, "%s %s", "touch", path);
    ret = system(sys_cmd);
    UNUSED(ret);
    // （1）创建唯一key值
    key = ftok(path,MAGIC_ID);
    if (key < 0)
    {
        perror("ftok err\n");
        return -1;
    }
    // （2）创建或打开共享内存
    shmid = shmget(key, shm_size, IPC_CREAT | 0666);
    if (shmid < 0)
    {

        perror("shmget err");
        return -1;
    }
    return 0;
}
// 获取共享内存地址
void *shm_getaddr()
{
    void *addr;
    addr = shmat(shmid, NULL, 0);
    if (addr == (void *)-1)
    {
        perror("shm_getaddr err");
        return NULL; // 或者根据需要处理错误
    }
    return addr;
}
// 建立数据库
int sqllite_init()
{
    db = NULL;
    // 1、打开/创建数据库
    int flag = sqlite3_open("./sb.db", &db);
    if (flag > 0)
    {
        printf("create sqlite err\n");
        return -1;
    }
    // 2.创建一个表
    char *errmsg = NULL; // 存储错误信息
    if (sqlite3_exec(db, "create table data(key int key,value string,time int);", NULL, NULL, &errmsg) != 0)
    {
        // printf("create table err:%s\n", errmsg);
    }
    return 0;
}
// 接收mqtt回调函数---指令处理
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char *payloadptr;
    payloadptr = message->payload;
    cJSON *msgroot = cJSON_Parse(payloadptr);
    if (msgroot == NULL)
    {
        printf("data err:%s\n", payloadptr);
    }
    else
    {
        // 判断指令
        cJSON *item = cJSON_GetObjectItem(msgroot, "type");
        if (item == NULL)
        {
            printf("no type\n");
        }
        printf("zltype%s:%d\n", item->string, item->valueint);
        zltype = item->valueint;
        // 采集指令---刷新上报
        if (zltype == 1)
        {
            cJSON *limit_cj = cJSON_GetObjectItem(msgroot, "limit");
            limit = limit_cj->valuestring;
            if (strcmp(limit, "part") == 0)
            {
                cJSON *keys = cJSON_GetObjectItem(msgroot, "keys");
                // 获取数组的大小
                int keys_num = cJSON_GetArraySize(keys);
                for (int i = 0; i < keys_num; i++)
                {
                    keys_part[i] = keys[i].valueint;
                }
            }
            // 遍历共享内存并上报
            periodic_reporting();
        }
        // 控制指令
        if (zltype == 2)
        {
            cJSON *data = cJSON_GetObjectItem(msgroot, "data");
            if (data == NULL)
            {
                printf("no data\n");
                return -1;
            }
            // 解析控制key、val，写入消息队列
            cJSON *key_ct = cJSON_GetObjectItem(data, "key");
            key_ctrl = key_ct->valueint;
            cJSON *val_ct = cJSON_GetObjectItem(data, "val");
            val_ctrl = val_ct->valuestring;
            char mtext[256] = {0};
            sprintf(mtext, "key:%d,val:%s\n", key_ctrl, val_ctrl);
            struct msgbuf msg_queue;
            // 读取共享内存内设备节点信息
            for (int i = 0; i < sum; i++)
            {
                if (shm_data[i].key == key_ctrl)
                {
                    if (shm_data[i].dev_type == 101) // stm32
                    {
                        strcpy(msg_queue.mdata,mtext);
                        msg_queue.mtype=101;
                        msg_queue_send(MSG_NAME,&msg_queue,sizeof(msg_queue),0);
                    }
                    else if (shm_data[i].dev_type == 102) // modbus
                    {
                        strcpy(msg_queue.mdata,mtext);
                        msg_queue.mtype=102;
                        msg_queue_send(MSG_NAME,&msg_queue,sizeof(msg_queue),0);
                    }
                }
            }
        }
        // 模式修改指令
        if (zltype == 3)
        {
            // 打开点表文件
            //   创建cJson节点
            int fd = open(NODE_PATH, O_RDONLY);
            if (fd < 0)
            {
                perror("open err");
                return -1;
            }
            char buf[4096] = {0};
            int ret = read(fd, buf, 1024);
            if (ret < 0)
            {
                perror("read error");
            }
            close(fd);
            cJSON *data = cJSON_GetObjectItem(msgroot, "data");
            if (data == NULL)
            {
                printf("no data\n");
                return -1;
            }
            item = cJSON_GetObjectItem(data, "type");
            if (item == NULL)
            {
                printf("no type\n");
                return -1;
            }
            sbtype = item->valueint;
            // 修改点表文件
            // 反序列化
            cJSON *node_file = cJSON_Parse(buf);
            // 获取report
            cJSON *report = cJSON_GetObjectItem(node_file, "report");
            // 创建int节点
            cJSON *report_type = cJSON_CreateNumber(sbtype);
            printf("sbtype:%d\n", sbtype);
            // 修改节点
            cJSON_ReplaceItemInObject(report, "type", report_type);
            item = cJSON_GetObjectItem(data, "period");
            if (item == NULL)
            {
                printf("no period\n");
                cJSON_DeleteItemFromObject(report, "period");
            }
            else
            {
                period = item->valueint;
                if (cJSON_GetObjectItem(report, "period") == NULL)
                {
                    cJSON_AddItemToObject(report, "period", cJSON_CreateNumber(period));
                }
                else
                {
                    // 创建int节点
                    cJSON *report_period = cJSON_CreateNumber(period);
                    // 修改节点
                    cJSON_ReplaceItemInObject(report, "period", report_period);
                }
            }
            // 序列化成字符串，写给文件
            char *newbuf = cJSON_Print(node_file);
            fd = open(NODE_PATH, O_RDWR | O_TRUNC);
            if (fd < 0)
            {
                perror("open err");
                return -1;
            }
            int ret_write = write(fd, newbuf, strlen(newbuf));
            if (ret == -1)
            {
                printf("write error\n");
            }

            close(fd); // 读取完后关闭文件
            cJSON_Delete(node_file);
            free(newbuf);
        }
        // 历史记录查询指令处理
        if (zltype == 4)
        {
            // 解析数据，并且存入全局变量
            cJSON *data = cJSON_GetObjectItem(msgroot, "data");
            cJSON *_key = cJSON_GetObjectItem(data, "key");
            key_num = _key->valueint;
            // 获取object大节点名字叫string的子节点
            _key = cJSON_GetObjectItem(data, "limit");
            cJSON *limit = cJSON_GetArrayItem(_key, 0);
            limit_head = limit->valueint;
            limit = cJSON_GetArrayItem(_key, 1);
            limit_tail = limit->valueint;
            printf("key:%d,limit_h:%d,limit_t:%d\n", key_num, limit_head, limit_tail);
            select_history(); // 查询指令
        }
    }
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    cJSON_Delete(msgroot);
    return 1;
}
// mqtt断开回调函数
void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}
// 连接mqtt服务器
void connect_to_mqtt_server()
{
    // 声明并初始化MQTT客户端连接选项结构体，使用初始值设定函数进行初始化
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    // 用于存储连接函数等操作的返回码
    int rc;
    // 声明并初始化MQTT消息结构体，同样使用初始值设定函数初始化
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    // 用于存储消息发布的交付令牌，可用于跟踪消息是否成功发布等操作
    MQTTClient_deliveryToken token;
    // 声明一个易失性的交付令牌变量，可能用于在异步操作等场景下跟踪交付状态
    // （例如在消息发布完成的回调函数中可能会更新该变量）
    volatile MQTTClient_deliveryToken deliveredtoken;
    // 创建MQTT客户端实例
    // 参数依次为：指向客户端对象的指针、MQTT服务器地址、客户端ID、持久化类型（这里是无持久化）、用户数据指针（这里为NULL）
    MQTTClient_create(&client, addr, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    // 设置连接选项中的保活时间间隔为20秒，保活机制用于维持客户端与服务器之间的连接状态
    conn_opts.keepAliveInterval = 20;
    // 设置连接选项中的清除会话标志为1，表示每次连接都开启一个新的会话，不恢复之前的会话状态
    conn_opts.cleansession = 1;
    // 设置回调函数
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, NULL);
    // 尝试连接到MQTT服务器
    // 如果连接失败，将返回码输出并以失败状态退出程序
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    // 订阅主题
    MQTTClient_subscribe(client, TOPIC, QOS);
}
// 轮寻共享内存并上报
int periodic_reporting()
{
    // 创建节点
    // 创建object主干节点
    cJSON *zqsbmessage = cJSON_CreateObject();
    cJSON *type = cJSON_CreateNumber(1);
    cJSON_AddItemToObject(zqsbmessage, "type", type);
    cJSON *result = cJSON_CreateNumber(0);
    cJSON_AddItemToObject(zqsbmessage, "result", result);
    // 创建数组节点
    cJSON *data = cJSON_CreateArray();
    cJSON *key = NULL;
    cJSON *val = NULL;
    int dataFound = 0; // 标志位
    // 轮询共享内存,并生成json
    for (int i = 0; i < sum; i++)
    {
        // 创建{}节点
        cJSON *Object = cJSON_CreateObject();
        key = cJSON_CreateNumber(shm_data[i].key);
        // 把item节点增加到object中作为子节点，item节点的键名为string
        cJSON_AddItemToObject(Object, "key", key);
        // 判断数据类型
        switch (shm_data[i].type)
        {
        case 1:
            val = cJSON_CreateNumber(shm_data[i].new_val.b_val);
            cJSON_AddItemToObject(Object, "val", val);
            dataFound = 1;
            break;
        case 2:
            val = cJSON_CreateString(i_to_str(shm_data[i].new_val.i_val));
            cJSON_AddItemToObject(Object, "val", val);
            dataFound = 1;
            break;
        case 3:
            val = cJSON_CreateString(f_to_str(shm_data[i].new_val.f_val));
            cJSON_AddItemToObject(Object, "val", val);
            dataFound = 1;
            break;
        default:
            break;
        }
        cJSON_AddItemToArray(data, Object);
    }
    cJSON_AddItemToObject(zqsbmessage, "data", data);

    if (!dataFound)
    {
        cJSON *result_err = cJSON_CreateNumber(1);
        cJSON_ReplaceItemInObject(zqsbmessage, "result", result_err);
    }
    char *buf = cJSON_Print(zqsbmessage);
    mqtt_send(buf);

    free(buf);
    cJSON_Delete(zqsbmessage);
}
// 变化上报
int change_reporting()
{
    // 创建节点
    // 创建object主干节点
    cJSON *bhsbmessage = cJSON_CreateObject();
    cJSON *type = cJSON_CreateNumber(1);
    cJSON_AddItemToObject(bhsbmessage, "type", type);

    cJSON *result = cJSON_CreateNumber(0);
    cJSON_AddItemToObject(bhsbmessage, "result", result);
    // 创建数组节点
    cJSON *data = cJSON_CreateArray();

    cJSON *key = NULL;
    cJSON *val = NULL;

    int data_changed = 0;
    // 轮寻共享内存,并生成json
    for (int i = 0; i < sum; i++)
    {
        // 创建{}节点
        cJSON *Object = cJSON_CreateObject();
        key = cJSON_CreateNumber(shm_data[i].key);
        // 把item节点增加到object中作为子节点，item节点的键名为string
        cJSON_AddItemToObject(Object, "key", key);
        int ret = memcmp(&(shm_data[i].new_val), &(shm_data[i].old_val), sizeof(union val_t));
        if (ret != 0)
        {
            data_changed = 1;
            switch (shm_data[i].type)
            {
            case 1:
                val = cJSON_CreateNumber(shm_data[i].new_val.b_val);
                cJSON_AddItemToObject(Object, "val", val);
                shm_data[i].old_val.b_val = shm_data[i].new_val.b_val;
                break;
            case 2:
                val = cJSON_CreateString(f_to_str(shm_data[i].new_val.i_val));
                cJSON_AddItemToObject(Object, "val", val);
                shm_data[i].old_val.i_val = shm_data[i].new_val.i_val;
                break;
            case 3:
                val = cJSON_CreateString(i_to_str(shm_data[i].new_val.f_val));
                cJSON_AddItemToObject(Object, "val", val);
                shm_data[i].old_val.f_val = shm_data[i].new_val.f_val;
                break;
            default:
                break;
            }
        }

        cJSON_AddItemToArray(data, Object);
    }
    if (data_changed)
    {
        cJSON_AddItemToObject(bhsbmessage, "data", data);
        char *buf = cJSON_Print(bhsbmessage);
        //发送mqtt资源
        mqtt_send(buf);
        // 释放资源
        free(buf);                 // 释放通过cJSON_Print生成的字符串
        cJSON_Delete(bhsbmessage); // 释放JSON对象
    }
    else
    {
        cJSON_Delete(bhsbmessage); // 如果没有数据变化，释放JSON对象
        cJSON_Delete(data);        // 如果data没有被使用也需要删除
    }
}
// 数据库存储
int data_storage()
{
    char *errmsg;
    // 论寻共享内存，放入数据库
    for (int i = 0; i < sum; i++)
    {
        char sql[128] = {0};
        int key = shm_data[i].key;
        char value[32] = {0};
        switch (shm_data[i].type)
        {
        case 1:
            strcpy(value, i_to_str(shm_data[i].old_val.b_val));
            break;
        case 2:
            strcpy(value, i_to_str(shm_data[i].old_val.i_val));
            break;
        case 3:
            strcpy(value, f_to_str(shm_data[i].old_val.f_val));
            break;
        default:
            break;
        }
        // 获取时间戳
        struct timeval tv;
        gettimeofday(&tv, NULL);
        sprintf(sql, "insert into data values(%d,\"%s\",%ld);", key, value, tv.tv_sec);
        if (sqlite3_exec(db, sql, NULL, NULL, &errmsg) != SQLITE_OK)
        {
            printf("insert err:%s\n", errmsg);
            return -1;
        }
    }
    return 0;
}
// 查询历史记录并发送给客户端
int select_history()
{
    // 检查指令类型
    if (zltype != 4)
    {
        return 1;
    }

    // 创建历史消息的JSON对象
    cJSON *history_message = cJSON_CreateObject();
    cJSON_AddNumberToObject(history_message, "type", 4);      // 指令类型：4 - 历史数据请求
    cJSON_AddNumberToObject(history_message, "result", 0);    // 成功返回0，失败返回1
    cJSON_AddNumberToObject(history_message, "key", key_num); // 设备的键值

    // 创建data数组，用于存放历史数据
    cJSON *data = cJSON_CreateArray();

    // 根据key值和时间范围去查询数据库
    char *errmsg;
    char **resultp = NULL;
    int nrow, ncolumn;
    int k = 1;           // 跳过表头，从第1行开始（nrow的第一行是列名）
    char sql[256] = {0}; // 增加sql长度以容纳时间范围
    sprintf(sql, "SELECT * FROM data WHERE key=%d AND time BETWEEN %d AND %d;", key_num, limit_head, limit_tail);

    // 执行查询
    sqlite3_get_table(db, sql, &resultp, &nrow, &ncolumn, &errmsg);
    // 判断查询是否成功
    if (errmsg != NULL)
    {
        printf("SQL Error: %s\n", errmsg);
        return 1;
    }

    // 遍历查询结果
    for (int i = 1; i <= nrow; i++)
    {
        // 从第1行开始读取数据，跳过表头
        cJSON *senddata = cJSON_CreateObject(); // 每一行数据创建一个JSON对象

        // 查询结果的列顺序是：key, value, time
        char *str_val = resultp[i * ncolumn + 1];  // 获取val字段 (第二列)
        char *str_time = resultp[i * ncolumn + 2]; // 获取time字段 (第三列)

        // 将时间戳和数据值添加到senddata对象
        cJSON_AddNumberToObject(senddata, "time", atoi(str_time)); // 时间戳
        cJSON_AddStringToObject(senddata, "val", str_val);         // 数据值（转换为浮动数值）

        // 将每一行的senddata添加到data数组中
        cJSON_AddItemToArray(data, senddata);
    }
    if (nrow == 0)
    {
        // 换成result 1
        // 替换某个节点
        //  创建一个新的节点
        cJSON *result_err = cJSON_CreateNumber(1);
        cJSON_ReplaceItemInObject(history_message, "result", result_err);
    }
    // 将data数组添加到history_message对象
    cJSON_AddItemToObject(history_message, "data", data);

    // 序列化JSON对象，使用无格式化的方式，节省内存
    char *buf = cJSON_PrintUnformatted(history_message);
    
    //发送mqtt
    mqtt_send(buf);

    // 释放内存
    free(buf);
    // 清理cJSON对象和数据库查询结果内存
    cJSON_Delete(history_message); // 释放cJSON对象
    return 0;
}
//mqtt发送函数
void mqtt_send(char *sendbuf)
{
    // 发送给broker服务器
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    pubmsg.payload = sendbuf;
    pubmsg.payloadlen = (int)strlen(sendbuf);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    MQTTClient_publishMessage(client, TOPIC_SEND, &pubmsg, &token);
}
//释放内存和消息队列
void delete_clear()
{
    //删除共享内存
    shmdt(shm_data);
    shmctl(shmid,IPC_RMID,NULL);
    //释放消息队列
    msgctl(msgid,IPC_RMID,NULL);
}

